package cn.sunline.dreamvalue.utils;

import cn.sunline.dreamvalue.controller.DreamValueController;
import cn.sunline.dreamvalue.dto.DreamValueDetailInDTO;
import cn.sunline.dreamvalue.dto.DreamAcctOpenDTO;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * @author 11292
 * @title: HbaseOprateUtil
 * @projectName dream-value
 * @description: hbase操作类
 * @date 2020/5/1210:25
 */
@Service
public class HbaseOprateUtil {

    public static final Logger log = LoggerFactory.getLogger(DreamValueController.class);
    @Autowired
    private HbaseEnv hbaseEnv;

    /**
     * 根据表的rowkey查询一条记录,返回infos中的列对应的值<br/>
     * @param tableName
     * 表名
     * @param rowkey
     * 拼接好的rowkey
     * @param infos key中保存family,value的list为column的列表
     * @return
     * 返回rowkey对应的这行数据,列及值用map存储
     */

    public  Map<String,String> query(String tableName, String rowkey, Map<String, List<String>> infos) throws IOException {

        Map<String, String> res ;

        Connection connection=getConnection();
        Table tab = connection.getTable(TableName.valueOf(tableName));

        try {
            res=innerQuery(tab,rowkey,infos);
        } finally {
            tab.close();
        }


        return res;
    }



    /**
     * 根据rowkey的列表,返回一组值<br/>
     * @param tableName
     * 查询的目标表
     * @param rowkeys
     * 要查询的rowkey清单
     * @param infos
     * 查询的列及管件信息
     * @return
     * 返回值的列表,每行值为一个map
     */

    public  List<Map<String,String>> query(String tableName,String[] rowkeys,Map<String,List<String>> infos) throws IOException {
        List<Map<String,String>> resList=new ArrayList<Map<String, String>>();

        Connection connection=getConnection();
        Table tab=connection.getTable(TableName.valueOf(tableName));
        try {
            for (String rowkey:rowkeys) {
                Map<String,String> res = innerQuery(tab, rowkey, infos);
                resList.add(res);
            }
        } finally {
            tab.close();
        }

        return resList;
    }



    /**
     * 执行更新
     * @param tableName     目标表
     * @param rowkey        行键
     * @param cols          列名和对应值
     * 格式为: <family:column,value>,<family:column,value><br/>
     * @return "success"
     * @throws IOException 内部异常
     */

    public  String update(String tableName, String rowkey, Map<String,String> cols) throws IOException{
        Connection connection = getConnection();
        Table tab = connection.getTable(TableName.valueOf(tableName));

        List<Put> arr = new ArrayList<Put>();
        for(String kv : cols.keySet()){
            Put put = new Put(Bytes.toBytes(rowkey));
            //<family:column,value>
            String[] tmp = kv.split(":");
            String family = tmp[0];
            String column = tmp[1];
            String value = cols.get(kv);
            put.addColumn(Bytes.toBytes(family), Bytes.toBytes(column), Bytes.toBytes(value));
            arr.add(put);
        }
        tab.put(arr);

        return "success";
    }



    /**
     * 执行删除
     * @param tableName     目标表
     * @param rowkeys        行键列表
     * @return "success"
     * @throws IOException 内部异常
     */

    public  String delete(String tableName, List<String> rowkeys) throws IOException{
        Connection connection = getConnection();
        Table tab = connection.getTable(TableName.valueOf(tableName));

        List<Delete> dels=new ArrayList<Delete>();

        for(String rowkey:rowkeys) {
            Delete del = new Delete(Bytes.toBytes(rowkey));
            dels.add(del);
        }
        tab.delete(dels);

        return "success";
    }



    /**
     * 根据rowkey的范围返回一组值<br/>
     * @param tableName
     * 查询的目标表
     * @param rowKeyFrom
     * rowkey的起始位置
     * @param rowKeyTo
     * rowkey的结束位置
     * @param infos
     * 查询的列等关键信息
     * @return
     * 返回数据的清单,每行为一个map
     * @throws IOException
     * 加载参数时报的异常
     */

    public  List<Map<String,String>> query(String tableName,String rowKeyFrom,String rowKeyTo,Map<String,List<String>> infos) throws IOException{
        List<Map<String,String>> resList=new ArrayList<Map<String, String>>();

        Connection connection=getConnection();
        Table tab=connection.getTable(TableName.valueOf(tableName));
        Scan scan=new Scan();
        scan.setMaxVersions(1);
        scan.setStartRow(Bytes.toBytes( rowKeyFrom));
        scan.setStopRow(Bytes.toBytes( rowKeyTo));
        for (String family : infos.keySet()) {
            for (String column : infos.get(family)) {
                scan.addColumn(Bytes.toBytes( family),Bytes.toBytes( column));
            }
        }
        try {
            ResultScanner rss = tab.getScanner(scan);
            for (Result rs : rss) {
                Cell[] cells = rs.rawCells();
                Map<String, String> res = transRowformCells(cells);
                resList.add(res);
            }
        } finally {
            tab.close();
        }
        return resList;
    }



    /**
     * 根据rowkey的范围返回一组值<br/>
     * @param tableName
     * 查询的目标表
     * @param rowKeyFrom
     * rowkey的起始位置
     * @param rowKeyTo
     * rowkey的结束位置
     * @param infos
     * 查询的列等关键信息
     * @param resultSize 查询结果最大行数
     * @return
     * 返回数据的清单,每行为一个map
     * @throws IOException
     * 加载参数时报的异常
     */

    public  List<Map<String,String>> query(String tableName,String rowKeyFrom,String rowKeyTo,Map<String,List<String>> infos,int resultSize) throws IOException{
        List<Map<String,String>> resList=new ArrayList<Map<String, String>>();

        Connection connection=getConnection();
        Table tab=connection.getTable(TableName.valueOf(tableName));
        Scan scan=new Scan();
        scan.setMaxVersions(1);
        scan.setStartRow(Bytes.toBytes( rowKeyFrom));
        scan.setStopRow(Bytes.toBytes( rowKeyTo));
        scan.setBatch(5000);
        int j = 0 ;
        for (String family : infos.keySet()) {
            for (String column : infos.get(family)) {
                scan.addColumn(Bytes.toBytes( family),Bytes.toBytes( column));
            }
        }
        try {
            ResultScanner rss = tab.getScanner(scan);
            for (Result rs : rss) {

                Cell[] cells = rs.rawCells();
                Map<String, String> res = transRowformCells(cells);
                resList.add(res);

                if(resultSize>0 && j>=resultSize){
                    break;
                }

                j++;
            }
        } finally {
            tab.close();
        }
        return resList;
    }

    /**
     * 根据rowkey的范围返回一组值倒叙排列<br/>
     * @param tableName
     * 查询的目标表
     * @param rowKeyFrom
     * rowkey的起始位置
     * @param rowKeyTo
     * rowkey的结束位置
     * @param infos
     * 查询的列等关键信息
     * @param resultSize 查询结果最大行数
     * @return
     * 返回数据的清单,每行为一个map
     * @throws IOException
     * 加载参数时报的异常
     */

    public  List<Map<String,String>> queryDesc(String tableName,String rowKeyFrom,String rowKeyTo,Map<String,List<String>> infos,int resultSize) throws IOException{
        List<Map<String,String>> resList=new ArrayList<Map<String, String>>();

        Connection connection=getConnection();
        Table tab=connection.getTable(TableName.valueOf(tableName));
        Scan scan=new Scan();
        scan.setMaxVersions(1);
        scan.setStartRow(Bytes.toBytes( rowKeyFrom));
        scan.setStopRow(Bytes.toBytes( rowKeyTo));
        scan.setBatch(5000);
        int j = 0 ;
        for (String family : infos.keySet()) {
            for (String column : infos.get(family)) {
                scan.addColumn(Bytes.toBytes( family),Bytes.toBytes( column));
            }
        }
        try {
            scan.setReversed(true);
            ResultScanner rss = tab.getScanner(scan);
            for (Result rs : rss) {

                Cell[] cells = rs.rawCells();
                Map<String, String> res = transRowformCells(cells);
                resList.add(res);

                if(resultSize>0 && j>=resultSize){
                    break;
                }

                j++;
            }
        } finally {
            tab.close();
        }
        return resList;
    }


    /**
     * 获取hbase连接
     * @return
     * 返回一个连接,此连接为线程安全的,多个请求复用
     * @throws IOException
     * 加载配置文件找不到时报异常
     */

    private  Connection getConnection() throws IOException {
        return hbaseEnv.getConnection();
    }


    /**
     * 根据表的rowkey查询一条记录,返回infos中的列对应的值<br/>
     * @param tab
     * hbase的表对象
     * @param rowkey
     * 行键
     * @param infos key中保存family,value的list为column的列表
     * @return
     * 返回行键对应的记录键值map
     */

    private  Map<String,String> innerQuery(Table tab, String rowkey, Map<String,List<String>> infos) throws IOException {

        Map<String, String> res ;

        Get get = new Get(rowkey.getBytes());
        for (String family : infos.keySet()) {
            for (String column : infos.get(family)) {
                get.addColumn(family.getBytes(), column.getBytes());
            }
        }

        Result rs = tab.get(get);
        Cell[] cells = rs.rawCells();
        res=transRowformCells(cells);

        return res;
    }


    /**
     　* @description: 分页查询梦想值流水
     　* @param [tableName, infos]
     　* @return org.apache.hadoop.hbase.client.ResultScanner
     　* @author 11292
     　* @date 2020/5/12 16:34
     */
    public ResultScanner queryDreamValuFlows(String tableName,DreamValueDetailInDTO infos) throws Exception{
        log.info("入参为：{}",infos.toString());
        Integer pagesize = 10;
        if(null != infos.getSize() && infos.getSize() > 0){
            pagesize = infos.getSize();
        }
        String dreamValuNo = infos.getDreamValueNo();
        String dreamValueNoType = infos.getType();
        Scan scan = new Scan();
        FilterList filterList = new FilterList(
                FilterList.Operator.MUST_PASS_ALL);
        //多取一条是因为第一条本身就是重复的需要剔除掉
        Filter pageFilter = new PageFilter(pagesize);


        //解决只带一个时间报错的问题
        DateFormat df = new SimpleDateFormat("yyyyMMdd");
        if(StringUtils.isEmpty(infos.getStartTime()) && !StringUtils.isEmpty(infos.getEndTime())){
            //只有结束时间，没有开始时间，初始化一个开始时间
            Date date = new Date(0);
            SingleColumnValueFilter scvf = new SingleColumnValueFilter(Bytes.toBytes("bi"), Bytes.toBytes("a2"), CompareFilter.CompareOp.GREATER_OR_EQUAL,
                    Bytes.toBytes(df.format(date)));
            filterList.addFilter(scvf);
        }

        if(StringUtils.isEmpty(infos.getEndTime()) && !StringUtils.isEmpty(infos.getStartTime())){
            //只有开始时间没有结束时间，初始化一个当前时间晚一天的结束时间
            Date date = new Date(System.currentTimeMillis()+1000*60*60*24);
            SingleColumnValueFilter scvf = new SingleColumnValueFilter(Bytes.toBytes("bi"), Bytes.toBytes("a2"), CompareFilter.CompareOp.LESS_OR_EQUAL,
                    Bytes.toBytes(df.format(date)));
            filterList.addFilter(scvf);
        }
        if (!StringUtils.isEmpty(infos.getStartTime())) {
            log.info("开始日期：{}",infos.getStartTime());
            SingleColumnValueFilter scvf = new SingleColumnValueFilter(Bytes.toBytes("bi"), Bytes.toBytes("a2"), CompareFilter.CompareOp.GREATER_OR_EQUAL,
                    Bytes.toBytes(infos.getStartTime()));
            filterList.addFilter(scvf);
        }

        if (!StringUtils.isEmpty(infos.getEndTime())) {
            log.info("结束日期：{}",infos.getEndTime());
            SingleColumnValueFilter scvf = new SingleColumnValueFilter(Bytes.toBytes("bi"), Bytes.toBytes("a2"), CompareFilter.CompareOp.LESS_OR_EQUAL,
                    Bytes.toBytes(infos.getEndTime()));
            filterList.addFilter(scvf);
        }

        if (!StringUtils.isEmpty(infos.getLastNo())) {
            log.info("最后一条流水号：{}",infos.getEndTime());
            SingleColumnValueFilter scvf = new SingleColumnValueFilter(Bytes.toBytes("bi"), Bytes.toBytes("a5"), CompareFilter.CompareOp.LESS,
                    Bytes.toBytes(infos.getLastNo()));
            filterList.addFilter(scvf);
        }

        Filter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL,
                new BinaryPrefixComparator((infos.getDreamValueNo()+infos.getType()).getBytes()));
        filterList.addFilter(rowFilter);
        filterList.addFilter(pageFilter);
        scan.setFilter(filterList);



        scan.setReversed(true);
        Connection connection=getConnection();
        Table tab=connection.getTable(TableName.valueOf(tableName));

        ResultScanner scanner = tab.getScanner(scan);

        return scanner;
    }

    /**
     　* @description: 分页查询客户完成度
     　* @param [tableName, infos]
     　* @return org.apache.hadoop.hbase.client.ResultScanner
     　* @author 82345
     　* @date 2020/5/14 12:14
     */
    public ResultScanner queryDreamValuTask(String tableName, DreamAcctOpenDTO infos) throws Exception{

        Integer pagesize = 10;
        String dreamValueNo = infos.getDreamValueNo();
        String dreamValueNoType = infos.getDreamAcctType();
        Scan scan = new Scan();
        FilterList filterList = new FilterList(
                FilterList.Operator.MUST_PASS_ALL);

        Filter pageFilter = new PageFilter(pagesize);

        if (!StringUtils.isEmpty(infos.getType())) {
            log.info("页签类型：{}",infos.getType());
            SingleColumnValueFilter scvf = new SingleColumnValueFilter(Bytes.toBytes("bi"), Bytes.toBytes("a2"), CompareFilter.CompareOp.EQUAL,
                    Bytes.toBytes(infos.getType()));
            filterList.addFilter(scvf);
        }
        Filter rowFilter = new RowFilter(CompareFilter.CompareOp.EQUAL,
                new BinaryPrefixComparator((infos.getDreamValueNo()+infos.getDreamAcctType()).getBytes()));
        filterList.addFilter(rowFilter);
        filterList.addFilter(pageFilter);
        scan.setFilter(filterList);
        scan.setReversed(true);

        Connection connection=getConnection();
        Table tab=connection.getTable(TableName.valueOf(tableName));

        ResultScanner scanner = tab.getScanner(scan);

        return scanner;
    }


    /**
     * 将一行数据保存各值的cell转换为一行数据的列和值的map<br/>
     * @param cells
     * 值列表
     * @return
     * 返回列和值的键值对
     * @throws UnsupportedEncodingException
     * 需要整体转为utf8编码
     */

    private  Map<String,String> transRowformCells(Cell[] cells) throws UnsupportedEncodingException {
        Map<String,String> res= new Hashtable<String, String>();

        if(null == cells || cells.length == 0){
            return res;
        }

        for (Cell cell : cells) {
            res.put(new String(CellUtil.cloneFamily(cell),"utf-8")+":"+ new String(CellUtil.cloneQualifier(cell),"utf-8"),new String(CellUtil.cloneValue(cell),"utf-8"));
        }
        return res;
    }

}
